package com.practice.car.cardataapp.schelduer.spark.task.sale;

import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.SaleRunner;
import com.practice.car.cardataapp.schelduer.spark.entityForSpark.CarMessage;
import com.practice.car.cardataapp.schelduer.spark.entityForSpark.ParseJson;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

public class CarPrice implements SaleRunner {
	public static Integer get(Double aDouble) {//这里传入汽车的价格，然后判断区间，给一个标记Integer
		if (aDouble < 10) {
			return 0;
		} else if (aDouble < 20) {
			return 1;
		} else if (aDouble < 30) {
			return 2;
		} else if (aDouble < 40) {
			return 3;
		} else if (aDouble < 50) {
			return 4;
		} else if (aDouble < 60) {
			return 5;
		} else
			return 6;
	}

	@Override
	public void run() {
		SparkConf conf = new SparkConf().setMaster(Schedule.sparkMaster).setAppName("MyApp");
		JavaSparkContext jsc = new JavaSparkContext(conf);
		//读取windows本地文件
		//JavaRDD<String> input = jsc.textFile("D:/style_items.json");
		//读取HDFS文件
		JavaRDD<String> input = jsc.textFile(Schedule.saleHdfsUrl);
		//数据写入RDD实体类
		JavaRDD<CarMessage> result = input.mapPartitions(new ParseJson());
		//sum[]保存各个价位的销量
		int[] sum = new int[7];
		//p[]保存各个价位销量占总销量比例
		double[] p = new double[7];
		int[] c = new int[7];
		//allcar保存总销量
		int allcar = 0;
		//初始化数据
		for (int i = 0; i < 7; i++) {
			sum[i] = 0;
			p[i] = 0;
			c[i] = i * 10 + 10;
		}
		int h = 0;
		double testsum = 0;
		JavaPairRDD<Integer, CarMessage> integerCarMessageJavaPairRDD = result.mapToPair(carmsg -> {
			return new Tuple2<>(get(carmsg.getPrice()), carmsg);
		});
		JavaPairRDD<Integer, CarMessage> sale = integerCarMessageJavaPairRDD.foldByKey(new CarMessage(), (carmsg1, carmsg2) -> {
			carmsg1.setSale(carmsg1.getSale() + carmsg2.getSale());
			return carmsg1;
		});

		List<CarMessage> carlist = sale.sortByKey().map(x -> x._2).collect();
		h = 0;
		testsum = 0;
		int ii = 0;
		for (CarMessage carMessage : carlist) {
			allcar += carMessage.getSale();
		}
		for (CarMessage carsale : carlist) {
			sum[ii] = carsale.getSale();
			h = (int) (carsale.getSale() * 1000.0 / allcar);
			p[ii] = h / 10.0;
			testsum += p[ii];
			ii++;
		}

		if ((100.0 - testsum) > 0) {
			//这样计算可以保留result_value小数点后四位,以此类推,1后面几个零就是保留小数点后几位数.
			//(double)(Math.round(result_value*10000)/10000.0)四舍五入
			p[6] = (double) (Math.round((100 - testsum) * 10) / 10.0);
		}

		//声明Connection对象
		Connection con = Schedule.getConnection();
		PreparedStatement psql;
		try {
			psql = con.prepareStatement("insert into car_price (idcar_price,min_price,max_price,sale,proportion) "
					+ " values(?,?,?,?,?)");
			for (int i = 0; i < 7; i++) {
				psql.setInt(1, (i + 1));
				psql.setDouble(2, (i * 10));
				psql.setDouble(3, ((i + 1) * 10));
				psql.setInt(4, sum[i]);
				psql.setDouble(5, p[i]);
				psql.executeUpdate();
			}
			con.close();
		} catch (SQLException e) {
			e.printStackTrace();
		}
		jsc.close();
	}
}
